/*
 *
 *  ------------------------------------------------------------------
 *  Copyright © 2017 Hangzhou DtDream Technology Co.,Lt d. All rights reserved.
 *  ------------------------------------------------------------------
 *        Product:  EMR
 *    Module Name: NEZHA
 *   Date Created: 17-10-23 下午4:22
 *    Description:
 *  ------------------------------------------------------------------
 *  Modification History
 *  DATE            Name           Description
 *  ------------------------------------------------------------------
 *  2017-10-23      NEZHA EMR
 *  ------------------------------------------------------------------
 * /
 */

package org.xukai.remoting.sdk.web.api;

import okhttp3.Credentials;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.ResponseBody;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.xukai.remoting.sdk.web.AccessKey;
import org.xukai.remoting.sdk.web.api.serde.CancelJobSerDe;
import org.xukai.remoting.sdk.web.api.serde.CreatePartitionSerDe;
import org.xukai.remoting.sdk.web.api.serde.GetDatabaseDetailSerDe;
import org.xukai.remoting.sdk.web.api.serde.GetJobReportSerDe;
import org.xukai.remoting.sdk.web.api.serde.GetStmtLogSerDe;
import org.xukai.remoting.sdk.web.api.serde.GetStmtResultSerDe;
import org.xukai.remoting.sdk.web.api.serde.GetStmtTrackerInfoSerDe;
import org.xukai.remoting.sdk.web.api.serde.GetTableNamesSerDe;
import org.xukai.remoting.sdk.web.api.serde.GetTableTypesSerDe;
import org.xukai.remoting.sdk.web.api.serde.GetTablesSerDe;
import org.xukai.remoting.sdk.web.api.serde.HttpSerDe;
import org.xukai.remoting.sdk.web.api.serde.SubmitJobSerDe;
import org.xukai.remoting.sdk.web.exception.NeZhaClientException;

/**
 * Created by moriarty on 17-2-23.
 */
public class RestAPIClient {

    private static final Logger LOG = LoggerFactory.getLogger(RestAPIClient.class);

    private OkHttpClient httpClient = new OkHttpClient.Builder().readTimeout(1, TimeUnit.MINUTES).build();
    private URL nezhaUrl;


    public RestAPIClient(URL nezhaUrl) {
        this.nezhaUrl = nezhaUrl;
    }

    public String submitJob(AccessKey accessKey, String project, List<String> sqls) throws NeZhaClientException {
        return execute(accessKey, "submitJob", new SubmitJobSerDe(sqls), project);
    }

    public SQLJobReport getJobReport(AccessKey accessKey, String project, String jobId) throws NeZhaClientException {
        return execute(accessKey, "getJobReport", new GetJobReportSerDe(), project, jobId);
    }

    public SQLJobStmtResult getJobStmtResult(AccessKey accessKey, String project, String jobId, int stmtIndex)
            throws NeZhaClientException {
        return execute(accessKey, "getJobStmtResult", new GetStmtResultSerDe(), project, jobId, stmtIndex);
    }

    public SQLJobStmtLog getJobStmtLog(AccessKey accessKey, String project, String jobId, int stmtIndex)
            throws NeZhaClientException {
        return execute(accessKey, "getJobStmtLog", new GetStmtLogSerDe(), project, jobId, stmtIndex);
    }

    public SQLTrackerInfo getJobStmtTrackerInfo(AccessKey accessKey, String project, String jobId, int stmtIndex)
            throws NeZhaClientException {
        return execute(accessKey, "getJobStmtTrackerInfo", new GetStmtTrackerInfoSerDe(), project, jobId,
                stmtIndex);
    }

    public void cancelJob(AccessKey accessKey, String project, String jobId) throws NeZhaClientException {
        execute(accessKey, "cancelJob", new CancelJobSerDe(), project, jobId);
    }

    @SuppressWarnings("unchecked")
    public List<String> getTableTypes(AccessKey accessKey, String project) throws NeZhaClientException {
        return execute(accessKey, "getTableTypes", new GetTableTypesSerDe(), project);
    }

    @SuppressWarnings("unchecked")
    public List<String> getTableNames(AccessKey accessKey, String project, String tablePattern, String[] tableTypes)
            throws NeZhaClientException {
        return execute(accessKey, "getTableNames", new GetTableNamesSerDe(), project,
                tablePattern == null ? "" : tablePattern, StringUtils.join(tableTypes, ","));
    }

    @SuppressWarnings("unchecked")
    @Deprecated
    public List<TableDesc> getTables(AccessKey accessKey, String project, String tablePattern, String[] tableTypes)
            throws NeZhaClientException {
        return execute(accessKey, "getTables", new GetTablesSerDe(), project,
                tablePattern == null ? "" : tablePattern, StringUtil.join(tableTypes, ","));
    }

    @SuppressWarnings("unchecked")
    public Page<TableDesc> getTablesPaged(AccessKey accessKey, String project, String tablePattern, String[] tableTypes,
                                          int pageNum, int pageSize) throws NeZhaClientException {
        return execute(accessKey, "getTablesPaged", new GetTablesPagedSerDe(), project,
                tablePattern == null ? "" : tablePattern, StringUtil.join(tableTypes, ","), pageNum, pageSize);
    }

    public TableDetail getTableDetail(AccessKey accessKey, String project, String table) throws NeZhaClientException {
        return execute(accessKey, "getTableDetail", new GetTableDetailSerDe(), project, table);
    }

    public SQLResult showPartitions(AccessKey accessKey, String project, String table, Map<String, String> partSpec)
            throws NeZhaClientException {
        try {
            if (partSpec == null) {
                partSpec = new HashMap<>();
            }
            return execute(accessKey, "showPartitions", new ShowPartitionsSerDe(), project, table,
                    JacksonCodec.serialize(partSpec));
        } catch (IOException e) {
            throw new NeZhaClientException("Failed to serialize partSpec", e);
        }
    }

    public DatabaseDetail getDatabaseDetail(AccessKey accessKey, String project) throws NeZhaClientException {
        return execute(accessKey, "getDatabaseDetail", new GetDatabaseDetailSerDe(), project);
    }

    public void createPartition(AccessKey accessKey, String dbName, String tableName, Map<String, String> partName) throws NeZhaClientException {
        execute(accessKey, "createPartition", new CreatePartitionSerDe(partName), dbName, tableName);
    }

    /* first argument has to be project name  */
    private <T> T execute(AccessKey accessKey, String method, HttpSerDe<T> serde, Object... args)
            throws NeZhaClientException {
        Map<String, String> headers = new HashMap<>();
        headers.put("Authorization", Credentials.basic(accessKey.id, accessKey.secret));
        headers.put("projectName", (String) args[0]);

        RestRequestBuilder builder = new RestRequestBuilder();
        try {
            builder.setArgs(args).setBody(serde.serialize()).setHeaders(headers)
                    .setRestAPI(RestAPIRegistry.lookupByMethodName(method)).setRootURL(nezhaUrl);
        } catch (IOException e) {
            throw new NeZhaClientException("Failed to serialize request body", e);
        }

        try {
            Request request = builder.build();
            LOG.debug("Execute " + request.toString());
            try (Response response = httpClient.newCall(request).execute()) {

                if (!response.isSuccessful()) {
                    throw new NeZhaClientException(response.code(), response.body().string());
                } else {
                    try (ResponseBody body = response.body()) {
                        return serde.deserialize(body);
                    } catch (IOException e) {
                        throw new NeZhaClientException("Failed to deserialize response body", e);
                    }
                }
            } catch (IOException e) {
                throw new NeZhaClientException("Can not connect to " + nezhaUrl, e);
            }
        } catch (MalformedURLException e) {
            throw new NeZhaClientException("Malformed request url", e);
        }
    }
}
